package com.mjf.spark.day04

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * 转换算子-foldByKey
 *    按照key对分区内以及分区间的数据进行处理
 *    -foldByKey是aggregateByKey简化版本
 *    -foldByKey(初始值)(分区内和分区间计算规则)
 */
object Spark05_Transformation_foldByKey {
  def main(args: Array[String]): Unit = {

    // 创建SparkConf配置文件
    val conf = new SparkConf().setMaster("local[*]").setAppName("Spark05_Transformation_foldByKey")
    // 创建SparkContext对象
    val sc = new SparkContext(conf)

    val rdd: RDD[(String, Int)] = sc.makeRDD(List(("a", 3), ("a", 2), ("c", 4), ("b", 3), ("c", 6), ("c", 8)), 2)

    // 如果分区内和分区间计算规则一样，并且不需要指定初始值，那么优先使用reduceByKey
//    rdd.reduceByKey(_ + _).collect().foreach(println)

    // 如果分区内和分区间计算规则不一样，并且需要指定初始值，那么优先使用aggregateByKey
//    val resRDD: RDD[(String, Int)] = rdd.aggregateByKey(0)(_ + _, _ + _)

    // 如果分区内和分区间计算规则一样，并且需要指定初始值，那么优先使用foldByKey
    val resRDD: RDD[(String, Int)] = rdd.foldByKey(0)(_ + _)

    resRDD.collect().foreach(println)

    // 关闭连接
    sc. stop()

  }
}

